package com.xiguaji.tiktok.task.awemecomment.service;

import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerMainRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerTypeRepository;
import com.gugee.tiktok.common.model.dto.SlimBloggerDto;
import com.gugee.tiktok.common.model.mysql.BloggerLog;
import com.gugee.tiktok.common.model.mysql.BloggerType;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.common.msg.CommentMessage;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xiguaji.tiktok.task.common.producer.AbstractProducerHandler;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-10-28 15:16
 */
@Service
@Slf4j
public class CommentService extends AbstractProducerHandler<CommentMessage> {


    @Autowired
    @Qualifier("kafkaTemplateComment")
    private KafkaTemplate<Long, CommentMessage> kafkaTemplateComment;
    @Autowired
    BloggerMainRepository bloggerMainRepository;
    @Autowired
    AppConfig appConfig;
    @Autowired
    AwemeRepository awemeRepository;
    @Autowired
    BloggerTypeRepository bloggerTypeRepository;
    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");

    private List<BloggerType> getAllBloggerTypes(){
        List<BloggerType> bloggerTypes = new ArrayList<>();
        BloggerType bloggerType = new BloggerType();
        bloggerType.setId(0);
        bloggerType.setName("ALL");
        bloggerTypes.add(bloggerType);
        bloggerTypes.addAll(bloggerTypeRepository.getBloggerTypeByIsDisplay(true));
        return bloggerTypes;
    }
    @Override
    public List<CommentMessage> findMessageObject(BloggerLog bloggerLog, TaskParam taskParam) {
        //查询播主排行中的视频，
        List<SlimBloggerDto> bloggers = bloggerMainRepository.findSlimBloggersFollowerAndCount(appConfig.getSysCollectFollowerCount(),((int)bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(),taskParam.getRecords());

        LocalDateTime localDateTime =  LocalDateTime.now().minusDays(appConfig.getLatestDays()+2);
        int condition = Integer.parseInt(localDateTime.format(formatter));
        long commentClTime = (System.currentTimeMillis()/1000-appConfig.getIntervalHours()*3600);
        List<CommentMessage> rst=new ArrayList<>(taskParam.getRecords()*50);

        bloggers.forEach(e->{
            //查询90天内视频列表，评论数未采集到要求数量,评论数增量要求,增量步长与最低评论数同值
            List<CommentMessage> tmp=  awemeRepository.findCommentMessagesByUidAndRecentTimes(e.getUid(),condition,appConfig.getCollectMaxComments(),appConfig.getCollectStepComments(),commentClTime);
            rst.addAll(tmp);
        });
        return rst;
    }
    @Override
    public void sendMsg(List<CommentMessage> messages,String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(appConfig.getRetryCommentTimes());
            msg.setRetryLimitTimes(appConfig.getRetryCommentTimes());
            kafkaTemplateComment.send(topic,msg);
        });
        XxlJobLogger.log("投递主题={},投递消息数={}", topic,messages.size());
        log.info("投递主题={},投递消息数={}", topic,messages.size());
    }

    public void sendMsgWithAllTypes(TaskParam taskParam) {

        LocalDateTime localDateTime =  LocalDateTime.now().minusDays(appConfig.getLatestDays()+1);
        int condition = Integer.parseInt(localDateTime.format(formatter));

        Set<Long> msgSended=new HashSet<>(taskParam.getRecords()*60);
        List<CommentMessage> rst=new ArrayList<>(taskParam.getRecords()*31);
        long commentClTime = (System.currentTimeMillis()/1000-appConfig.getIntervalHours()*3600);
        getAllBloggerTypes().forEach(b->{
        //查询各个分类的topN
            List<CommentMessage> messages2=awemeRepository.getCommentsRankingOrderByScore(condition,b.getName().equals("ALL")?"":b.getName(),appConfig.getCollectMaxComments(),appConfig.getCollectStepComments(),0, taskParam.getRecords(),commentClTime);
            messages2.forEach(commentMessage->{
                if(!msgSended.contains(commentMessage.getOAwemeId())) {
                    msgSended.add(commentMessage.getOAwemeId());
                    commentMessage.setRetryTimes(appConfig.getRetryCommentTopTimes());
                    kafkaTemplateComment.send(taskParam.getTopic(),commentMessage);
                }
            });

            List<BloggerMessage> messages=bloggerMainRepository.findBloggerMessagesByBloggerTypeSortDesc(appConfig.getSysCollectFollowerCount(),b.getName().equals("ALL")?"":b.getName(), taskParam.getRecords(),0);
            messages.forEach(msg->{
                try {
                    List<CommentMessage> tmp = awemeRepository.findCommentMessagesByUidAndRecentTimes(msg.getUid(), condition, appConfig.getCollectMaxComments(), appConfig.getCollectStepComments(), commentClTime);
                    tmp.forEach(commentMessage -> {
                        if (!msgSended.contains(commentMessage.getOAwemeId())) {
//                        msgSended.add(commentMessage.getOAwemeId());
                            commentMessage.setRetryTimes(appConfig.getRetryCommentTopTimes());
                            kafkaTemplateComment.send(taskParam.getTopic(), commentMessage);
                        }
                    });
                }catch (Exception e){
                    XxlJobLogger.log("error:Uid={},minTime={},MaxComments={},StepComments={},commentClTime={}", msg.getUid(),condition, appConfig.getCollectMaxComments(), appConfig.getCollectStepComments(), commentClTime);
                }

            });



            XxlJobLogger.log("排行版播主类别={},消息数: {}", b.getName().equals("ALL")?"":b.getName(),msgSended.size());
        });
        XxlJobLogger.log("构造消息总数: {}",  msgSended.size());

    }
}
